package com.atguigu.kafka.pro.question;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.TopicPartition;

import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author ：剧情再美终是戏
 * @description：模拟kafka消费者能力不足情况
 * @mail : 13286520398@163.com
 * @date ：Created in 2020/1/24 19:13
 * @modified By：
 * @version: 1.0$
 */
public class OneToManyConsumerDmo {

    private static Deque<Map<String, Object>> linkedList = new LinkedList<>();
    private static Deque<Map<String, Object>> errorLinkedList = new LinkedList<>();

    private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 5,
            10000, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());

    private static long maxOffset = 0;

    private synchronized static void setMaxOffset(long offset) {
        if (offset > maxOffset) {
            maxOffset = offset;
        }
    }

    public static void main(String[] args) {

        // 创建一个消费者，用户消费kafka数据
        consumer(Constan.topic);

        // 创建一个异常入库线程，将异常信息入库
        insertDBErrorThreads();

        // 创建一个入库线程池，用户将数据存入DB
        insertDBThreads();
    }


    // 创建一个消费者，用户消费kafka数据
    public static void consumer(String topic) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop101:9092,hadoop102:9092,hadoop103:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test01");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        Consumer<Integer, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(topic));
        Thread consucmerThread = new Thread(() -> {
            while (true) {
                ConsumerRecords<Integer, String> records = consumer.poll(Constan.bathSize);
                List<ConsumerBean> list = new ArrayList<>();
                for (ConsumerRecord<Integer, String> record : records) {
                    String[] values = record.value().split("-");
                    ConsumerBean consumerBean = new ConsumerBean(record.offset(), values[0], Long.parseLong(values[1]));
                    list.add(consumerBean);
                }
                if (list.size() > 0) {
                    Map<String, Object> map = new HashMap<>();
                    map.put("data", list);
                    map.put("errorCount", 0);
                    linkedList.add(map);
                }
                // 提交offset
                if (maxOffset != 0) {
                    Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
                    offsets.put(new TopicPartition(Constan.topic, 0), new OffsetAndMetadata(maxOffset, null));
                    consumer.commitSync(offsets);
                }
            }
        });
        consucmerThread.start();
    }

    // 创建一个入库线程池，用户将数据存入DB
    public static void insertDBThreads() {
        while (true) {
            threadPoolExecutor.execute(() -> {
                Map<String, Object> poll = null;
                try {
                    poll = linkedList.poll();

                    if (null == poll || poll.size() == 0) {
                        return;
                    }

                    if (new Random().nextBoolean()) {
                        throw new RuntimeException("new Random().nextBoolean() error");
                    }
                } catch (Exception e) {
                    if (null != poll) {
                        int errorCount = Integer.parseInt(poll.get("errorCount").toString());
                        // TODO 如果错误次数到了指定次数，默认为3次，就不做重试操作了，将其放到错误队列中
                        if (errorCount >= Constan.errorCount) {
                            poll.put("msg", e);
                            errorLinkedList.add(poll);
                        } else {
                            // TODO 这里如果发生异常，就将数据放回队列中，并将错误次数+1
                            poll.put("errorCount", errorCount + 1);
                            linkedList.addFirst(poll);
                        }
                    }
                } finally {
                    // TODO 入库，提交事务
                    try {
                        Thread.sleep(1000);

                        // 记录最大的offset
                        if (null != poll) {
                            for (ConsumerBean consumerBean : (List<ConsumerBean>) poll.get("data")) {
                                setMaxOffset(consumerBean.getOffset());
                            }
                            System.out.println(maxOffset);
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
    }

    // 创建一个入库线程池，用户将数据存入DB
    public static void insertDBErrorThreads() {
        new Thread(() -> {
            while (true) {
                System.out.println("errorLinkedList--------------------------------" + errorLinkedList.size());
                Map<String, Object> poll = errorLinkedList.poll();

                if (null == poll || poll.size() == 0) {
                    continue;
                }
                // TODO 异常信息入库,如果入库时有异常，重新放回异常队列中，如果异常队列长度达到阀值，预警操作
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    private static class ConsumerBean {
        private long offset;

        private String val;

        private long ts;

        public long getOffset() {
            return offset;
        }

        public String getVal() {
            return val;
        }

        public long getTs() {
            return ts;
        }

        public ConsumerBean() {
        }

        public ConsumerBean(long offset, String val, long ts) {
            this.offset = offset;
            this.val = val;
            this.ts = ts;
        }

        @Override
        public String toString() {
            return "ConsumerBean{" +
                    "offset=" + offset +
                    ", val='" + val + '\'' +
                    ", ts=" + ts +
                    '}';
        }
    }
}
